package com.qiangesoft.rabbitmq.consumer;

import com.qiangesoft.rabbitmq.producer.MessageConfig;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 消费者
 *
 * @author qiangesoft
 * @date 2024-05-08
 */
@Slf4j
@Component
public class ConsumerListener {

    private AtomicInteger atomicInteger = new AtomicInteger(0);

    @RabbitListener(queues = MessageConfig.QUEUE, ackMode = "AUTO")
    public void autoAck(Message message) throws InterruptedException {
        // 模拟业务处理
        Thread.sleep(200);

        // todo messageId幂等性处理
        String messageId = message.getMessageProperties().getMessageId();
        log.info("消费者auto接收到消息，messageId：{}，message：{}", messageId, new String(message.getBody(), StandardCharsets.UTF_8));

        try {
            // 模拟出错
            if (atomicInteger.incrementAndGet() % 3 == 0) {
                System.out.println(1 / 0);
            }
            // todo 业务处理
            log.info("消费者auto消息处理完成");
        } catch (Exception e) {
            e.printStackTrace();
            throw e;
        }
    }

    @RabbitListener(queues = MessageConfig.QUEUE, ackMode = "MANUAL")
    public void manualAck(Message message, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException, InterruptedException {
        // 模拟业务处理
        Thread.sleep(200);

        // todo messageId幂等性处理
        String messageId = message.getMessageProperties().getMessageId();
        byte[] body = message.getBody();
        log.info("消费者manual接收到消息，messageId：{}，message：{}", messageId, new String(body, StandardCharsets.UTF_8));

        try {
            // 模拟出错
            if (atomicInteger.incrementAndGet() % 3 == 0) {
                System.out.println(1 / 0);
            }
            // todo 业务处理
            channel.basicAck(deliveryTag, true);
            log.info("消费者manual消息处理完成");
        } catch (Exception e) {
            e.printStackTrace();
            channel.basicNack(deliveryTag, false, true);
        }
    }

}